Kafka-clients: records(TopicPartition): unit tests + some (potential) improvements#924
Conversation
from original
merge from original
…mentation. introduce safeguards when instrumenting iterator() so that duplication of traces is not allowed. do not allow subList() to be instrumented
mar-kolya
left a comment
mar-kolya
left a comment
There was a problem hiding this comment.
Thank you so much for working on this!
Just couple of minor changes requested.
Thanks!
| @Override | ||
| public Iterator<ConsumerRecord> iterator() { | ||
| return super.iterator(); | ||
| } |
There was a problem hiding this comment.
Minor note: I think you can drop this method from here - it will be inherited from iterable
There was a problem hiding this comment.
Yep, I understand, but I was getting compiler warnings regarding @NotNull annotation.. not a big deal ?
There was a problem hiding this comment.
No. you can leave this as is if you prefer
There was a problem hiding this comment.
I will take it out, we can avoid a call in the stack. thanks!
| cleanup: | ||
| producerFactory.stop() | ||
| container?.stop() | ||
| embeddedKafka.after() |
There was a problem hiding this comment.
It looks like you could make kafka rule use @Rule instead of @Shared @ClassRule and then you would not need to manually call before/after on it
| producer.send(new ProducerRecord<Integer, String>(SHARED_TOPIC, kafkaPartition, null, greeting)) | ||
| TEST_WRITER.waitForTraces(1) | ||
|
|
||
| then: |
There was a problem hiding this comment.
Looks like you may be able to simplify this to something like:
when:
def greeting = "Hello from MockConsumer!"
producer.send(new ProducerRecord<Integer, String>(SHARED_TOPIC, kafkaPartition, null, greeting))
then:
TEST_WRITER.waitForTraces(1) // ensure consistent ordering of traces
def pollResult = KafkaTestUtils.getRecords(consumer)
def records = pollResult.records(new TopicPartition(SHARED_TOPIC, kafkaPartition)).iterator()
def first
while (records.hasNext()) {
first = records.next()
break
}
records.hasNext() == false
first.value() == greeting
first.key() == null
There was a problem hiding this comment.
I will change the WAIT_TRACES of producer to the then block:
On other note, I don't need to wait for traces when consuming?
There was a problem hiding this comment.
assertTraces(2) { waits to see exactly two traces. If I understand things correctly you need TEST_WRITER.waitForTraces(1) only once right before consuming - to make sure producer traces have been shipped - so two traces do not appear out of order at assertion time.
There was a problem hiding this comment.
yep makes sense. I have updated the PR as per your comments
| import (test),org.junit,EPL-1.0,Copyright © 2002-2017 JUnit. All Rights Reserved. | ||
| import (test),org.assertj,Apache-2.0,Copyright 2012-2017 the original author or authors. | ||
| import (test),org.mockito,MIT,Copyright (c) 2007 Mockito contributors | ||
| kafka-clients: method records(TopicPartition) partial+test,"Aspect Software, Inc.", Apache-2.0, "Copyright (C) Aspect Software, Inc." |
There was a problem hiding this comment.
I do not think this change is needed. I think this file is specifically for external deps that we ship with the built agent. This change doesn't add new dependencies.
There was a problem hiding this comment.
sure, this is more about the contribution as I/we thought it was what this is file is for. Is there anyway of the contribution being recognised in some other place? Thanks!
There was a problem hiding this comment.
We will certainly give you a shout-out in our release notes. Thanks again for taking the time and effort to contribute.
There was a problem hiding this comment.
Appreciate it! Happy to contribute!
… refactor unit tests to expect only one element to be consumed. Kafka embedded instance as a Rule
tylerbenson
added
tag: community
3 participants
I think these instrumented methods should safeguard for misuse and therefore prevent inaccurate traces. Since these methods can only be instrumented when applications calls them as it is imposed by the synchronous nature of the Kafka-client, then we should ensure (up to certain point) the agent collects what exactly the Kafka-client has processed. One example is, for instance, the application calls the iterator multiple times.
subList should not be instrumented as the application can remove elements of the list (as the method implies) and therefore, again, wrong traces can be collected which won't be the real representation of what has been received by the client